package org.alis.rpc.core;

import com.google.common.collect.Maps;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.alis.rpc.common.util.ServerUtils;
import org.alis.rpc.common.util.ThreadPoolUtils;
import org.alis.rpc.registry.ServerRegistry;

import java.util.Map;
import java.util.Objects;

/**
 * NettyServer desc: rpc 服务端核心启动方法
 *
 * @author luochuan
 * @date 2020/9/21
 */
@Slf4j
public class NettyServer implements Server {


    private String serverAddress;
    private ServerRegistry serverRegistry;
    private Map<String, Object> serviceMap = Maps.newHashMap();

    public NettyServer(String serverAddress, String registryAddress) {
        this.serverAddress = serverAddress;
        this.serverRegistry = new ServerRegistry(registryAddress);
    }

    public void addService(String interfaceName, String version, Object serviceBean) {
        log.info("Adding service, interface: {}, version: {}, bean：{}", interfaceName, version, serviceBean);
        String serviceKey = ServerUtils.makeServiceKey(interfaceName, version);
        serviceMap.put(serviceKey, serviceBean);
    }

    @Override
    public void start() {
        ThreadPoolUtils.execute(()->{
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                        .childHandler(new RpcServerInitializer(serviceMap))
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);

                String[] array = serverAddress.split(":");
                String host = array[0];
                int port = Integer.parseInt(array[1]);
                ChannelFuture future = bootstrap.bind(host, port).sync();

                if (serverRegistry != null) {
                    serverRegistry.registerService(host, port, serviceMap);
                }
                log.info("Server started on port {}", port);
                future.channel().closeFuture().sync();
            } catch (Exception e) {
                if (e instanceof InterruptedException) {
                    log.info("Rpc server remoting server stop");
                } else {
                    log.error("Rpc server remoting server error", e);
                }
            } finally {
                try {
                    Objects.requireNonNull(serverRegistry).unregisterService();
                    workerGroup.shutdownGracefully();
                    bossGroup.shutdownGracefully();
                } catch (Exception ex) {
                    log.error(ex.getMessage(), ex);
                }
            }
        });
    }

    @Override
    public void stop() {
        ThreadPoolUtils.shutdown();
    }
}
